package com.yyx.up.consumer.dao;

import com.yyx.up.common.bean.BaseDao;
import com.yyx.up.common.constant.Names;
import com.yyx.up.common.constant.ValueConstant;
import com.yyx.up.consumer.bean.WatchLog;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.List;

/**
 * Description: Hbase数据访问对象
 * @Author: yyx
 * @Create: 2023/4/14 16:27
 * @Version: 8
 */
public class HBaseDao extends BaseDao {
    /**
     * 初始化
     */
    public void init() throws Exception{
        start();

        createNamespaceNX(Names.NAMESPACE.getValue());//有则什么都不做
        createTableXX(Names.TABLE.getValue(),"com.yyx.up.consumer.coprocessor.InsertFilm_userCoprocessor", ValueConstant.REGION_COUNT, Names.UP_USER_Film.getValue(), Names.UP_Film_USER.getValue());//有则覆盖掉

        end();
    }

    /**
     * 插入对象
     * @param watchLog
     * @throws Exception
     */
    public void insertData(WatchLog watchLog) throws Exception{
        watchLog.setRowkey(genRegionNum(watchLog.getTel(),
                watchLog.getWatchtime()) + "_" + watchLog.getTel() + "_" + watchLog.getWatchtime() +
                        "_" + watchLog.getFilm() + "_" + watchLog.getDuration());
        putData(watchLog);
    }

    /**
     * 插入数据
     * @param value
     */
    public void insertData(String value) throws Exception {
        // 将观影日志保存到Hbase表中
        // 1.获取观影日志数据
        String[] values = value.split("\t");
        String tel = values[0];
        String film = values[1];
        String watchtime = values[2];
        String duration = values[3];
        // 2.创建数据对象
            // rowkey设计
            // 1) 长度原则：最大值64KB，推荐长度10~100byte；最好是8的倍数，能短则短，rowkey太长会影响性能
            // 2) 唯一原则
            // 3) 散列原则：
            //   3-1)盐值散列：不能使用时间戳直接作为rowkey；在rowkey前增加随机数
            //   3-2)字符串反转：前面有规律，后面没规律（时间戳、电话号码）
            //   3-3)计算分区号：hashMap
            // rowkey = regionNum + tel + watchtime + film + duration
        // 用户观影数据
        String rowkey = genRegionNum(tel,watchtime) + "_" + tel + "_" + watchtime + "_" + film + "_" + duration + "_1";
        Put put = new Put(Bytes.toBytes(rowkey));
        byte[] family = Bytes.toBytes(Names.UP_USER_Film.getValue());
        put.addColumn(family,Bytes.toBytes("tel"),Bytes.toBytes(tel));
        put.addColumn(family,Bytes.toBytes("film"),Bytes.toBytes(film));
        put.addColumn(family,Bytes.toBytes("watchtime"),Bytes.toBytes(watchtime));
        put.addColumn(family,Bytes.toBytes("duration"),Bytes.toBytes(duration));
        put.addColumn(family,Bytes.toBytes("flg"),Bytes.toBytes("1"));
        // 电影被观看数据
        /*String filmRowkey = genRegionNum(film,watchtime) + "_" + film + "_" + watchtime + "_" +
                tel + "_" + duration + "_0";
        Put filmPut = new Put(Bytes.toBytes(filmRowkey));
        byte[] filmFamily = Bytes.toBytes(Names.UP_Film_USER.getValue());
        filmPut.addColumn(filmFamily,Bytes.toBytes("film"),Bytes.toBytes(film));
        filmPut.addColumn(filmFamily,Bytes.toBytes("tel"),Bytes.toBytes(tel));
        filmPut.addColumn(filmFamily,Bytes.toBytes("watchtime"),Bytes.toBytes(watchtime));
        filmPut.addColumn(filmFamily,Bytes.toBytes("duration"),Bytes.toBytes(duration));
        filmPut.addColumn(filmFamily,Bytes.toBytes("flg"),Bytes.toBytes("0"));*/
        // 3.保存数据
        List<Put> puts = new ArrayList<Put>();
        puts.add(put);
        //puts.add(filmPut);
        putData(Names.TABLE.getValue(),puts);
    }
}
